package data.platform.merge.event;

import data.platform.common.event.CacheLoadedEvent;
import data.platform.common.event.MetricValueEvent;
import data.platform.merge.config.MetricMergeConfig;
import data.platform.merge.internal.cache.MergeItemCache;
import data.platform.merge.internal.command.MergeService;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Component;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import javax.annotation.PostConstruct;

@Component
@Slf4j
@RequiredArgsConstructor
public class MetricEventMergeProcess implements ApplicationListener<CacheLoadedEvent> {

    final ApplicationContext applicationContext;

    final MetricMergeConfig metricMergeConfig;

    final MeterRegistry dataSetMeterRegistry;

    final MergeItemCache mergeItemCache;

    final MergeService mergeService;

    Counter saveCounter;

    Counter dropCounter;

    private volatile boolean cacheLoaded = false;

    @PostConstruct
    public void init() {

        saveCounter = Counter.builder("dp.metricMerge.count.save")
                .register(dataSetMeterRegistry);

        dropCounter = Counter.builder("dp.metricMerge.count.drop")
                .register(dataSetMeterRegistry);

        ConfigurableApplicationContext context = (ConfigurableApplicationContext) applicationContext;
        Flux<MetricValueEvent> metricValueEventFlux = Flux.create(fluxSink -> {
            context.addApplicationListener((ApplicationListener<MetricValueEvent>) metricValueEvent -> {
                if (cacheLoaded) {
                    fluxSink.next(metricValueEvent);
                }
            });
        });

        bizProcessPipe(metricValueEventFlux).subscribe();
    }

    private Flux<Integer> bizProcessPipe(Flux<MetricValueEvent> metricValueEventFlux) {
        // -Dreactor.bufferSize.small=20 这个值设置了生产者生产数据后reactor的队列容量上限，达到后执行overflow操作，默认值为256
        // 首先使用onBackpressureBuffer容量，达到上限后，使用reactor.bufferSize.small容量
        // 总的缓存是onBackpressureBuffer+reactor.bufferSize.small
        return metricValueEventFlux
                // maxSize不是队列的长度，是reactor把数据发送消费者前缓存的数据容量
                .onBackpressureBuffer(metricMergeConfig.getBufferSize(),
                        metricEvent -> {
                            log.info("merge onBackpressureBuffer.....................");
                            dropCounter.increment();
                        },
                        BufferOverflowStrategy.DROP_OLDEST)
                .map(MetricValueEvent::getMetricValues)
                .flatMap(Flux::fromIterable)
                .filter(metricValue -> metricValue.getTag().containsKey("moc"))
                .filter(metricValue -> mergeItemCache.isMerge(metricValue.getMetric(), metricValue.getTag().get("moc")))
                .publishOn(Schedulers.parallel())
                .flatMap(metricValue -> mergeService.process(metricValue))
                .onErrorContinue((throwable, o) -> log.error("{}, continuing", throwable.getMessage()))
                .doOnNext(count -> saveCounter.increment(count));
    }

    @Override
    public void onApplicationEvent(CacheLoadedEvent event) {
        log.info("cache load finish, start aggregating......");
        cacheLoaded = true;
    }
}
